/*
 * Copyright (c) 2020 Carnegie Lee
 * All rights reserved
 */
package com.geekc.kafka.config;

import com.geekc.kafka.constant.KafkaConst;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.*;
import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
import org.springframework.kafka.listener.ContainerProperties;
import org.springframework.kafka.requestreply.ReplyingKafkaTemplate;

import java.util.HashMap;
import java.util.Map;

/**
 * KafkaReplyingTemplate配置类
 * @author Carnegie Lee
 */
@Configuration
public class KafkaReplyingTemplateConfig {

    /** ReplyingKafkaTemplate */
    @Bean
    public ReplyingKafkaTemplate replyKafkaTemplate(ProducerFactory producerFactory, ConcurrentMessageListenerContainer repliesContainer) {
        return new ReplyingKafkaTemplate<>(producerFactory, repliesContainer);
    }

    /** 监听器容器 */
    @Bean
    public ConcurrentMessageListenerContainer repliesContainer(ConcurrentKafkaListenerContainerFactory kafkaListenerContainerFactory) {
        ConcurrentMessageListenerContainer repliesContainer = (ConcurrentMessageListenerContainer) kafkaListenerContainerFactory.createContainer(KafkaConst.KAFKA_REPLYING_TOPIC);
        repliesContainer.getContainerProperties().setGroupId(KafkaConst.KAFKA_REPLYING_GROUP);
        repliesContainer.setAutoStartup(false);
        return repliesContainer;
    }

    /** 并发监听器容器工厂 */
    @Bean
    public ConcurrentKafkaListenerContainerFactory kafkaListenerContainerFactory(ConsumerFactory consumerFactory, ProducerFactory producerFactory) {
        ConcurrentKafkaListenerContainerFactory factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory);
        factory.setReplyTemplate(new KafkaTemplate(producerFactory));
        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
        return factory;
    }

    /** 响应生产者工厂 */
    @Bean
    public ProducerFactory producerFactory(KafkaProperties kafkaProperties) {
        Map props = new HashMap<>();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaProperties.getBootstrapServers());
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, kafkaProperties.getProducer().getKeySerializer());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, kafkaProperties.getProducer().getValueSerializer());
        return new DefaultKafkaProducerFactory(props);
    }

    /** 响应消费者工厂 */
    @Bean
    public ConsumerFactory consumerFactory(KafkaProperties kafkaProperties) {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaProperties.getBootstrapServers());
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, kafkaProperties.getConsumer().getKeyDeserializer());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, kafkaProperties.getConsumer().getValueDeserializer());
        return new DefaultKafkaConsumerFactory(props);
    }
}
